package cn.shuangbofu.nocturne.server.dispatcher;

import cn.shuangbofu.nocturne.core.entity.ExecutorEntry;
import cn.shuangbofu.nocturne.message.ResponseMessage;
import cn.shuangbofu.nocturne.protobuf.TaskProto;
import cn.shuangbofu.nocturne.protobuf.utils.RequestFactory;
import com.google.common.collect.Lists;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * Created by shuangbofu on 2020/7/25 下午8:11
 */
public class MockDispatcher {

    protected static final Logger LOGGER = LoggerFactory.getLogger(MockDispatcher.class);

    ArrayBlockingQueue<MockTask> queue = new ArrayBlockingQueue<>(3000);

    public void start() {
        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        service.scheduleWithFixedDelay(() -> {
            MockTask task = new MockTask();
            LOGGER.info("待执行任务「{}」进入队列", task.getId());
            put2Queue(task);
            LOGGER.info("当前所有任务[{}]", Lists.newArrayList(queue).stream().map(i -> String.valueOf(i.getId())).collect(Collectors.joining(",")));
        }, 10, 5, TimeUnit.SECONDS);

        ExecutorService executor = Executors.newSingleThreadExecutor();
        executor.submit(new DispatcherThread());
    }

    private void put2Queue(MockTask task) {
        try {
            queue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Data
    static class MockTask {
        static AtomicInteger number = new AtomicInteger();
        long id;
        String content;

        public MockTask() {
            id = number.incrementAndGet();
            content = UUID.randomUUID().toString();
        }
    }

    class DispatcherThread extends Thread {
        @Override
        public void run() {
            while (true) {
                try {
                    MockTask task = null;
                    try {
                        task = queue.take();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }

                    if (task == null) {
                        continue;
                    }
                    LOGGER.info("任务「{}」出队列分发", task.getId());
                    ExecutorSelector selector = RandomExecutorSelector.getInstance();
                    ExecutorEntry executorEntry = selector.select(1);

                    if (executorEntry == null) {
                        put2Queue(task);
                        LOGGER.warn("没有找到可用的执行服务器！任务「{}」重新进入队列", task.getId());
                        continue;
                    }

                    LOGGER.info("选择执行服务器 「{}」", executorEntry);
                    ResponseMessage response = executorEntry.getChannel().request(RequestFactory.submitTaskRequest(task.getId(), task.getContent()));

                    if (!response.isSuccess()) {
                        LOGGER.error("异常:{}", response.getMessage());
                        put2Queue(task);
                        LOGGER.info("任务「{}」出错，重新入队列", task.getId());
                        continue;
                    }

                    TaskProto.SubmitTaskResponse res = response.getData(TaskProto.SubmitTaskResponse.class);

                    if (!res.getAccept()) {
                        put2Queue(task);
                        LOGGER.warn("任务「{}」被拒绝，重新进入队列", task.getId());
                        continue;
                    }

                    LOGGER.info("任务「{}」接收成功！", task.getId());
                    // TODO
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
